package org.example.kafka.rebalance;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import org.example.kafka.common.CommonConstant;
import org.example.kafka.common.KafkaConstants;
import org.example.vo.DemoUser;

import java.util.concurrent.*;

/**
 *
 */
public class RebalanceProducer {
    private static final int MSG_SIZE = 60;
    private static final int THREAD_POOL_SIZE = 10;
    private static CountDownLatch countDownLatch
            = new CountDownLatch(MSG_SIZE);
    /**
     * 使用 ThreadFactoryBuilder 创建自定义线程名称的 ThreadFactory
     */
    private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
            .setNameFormat("hyn-demo-pool-%d").build();
    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(THREAD_POOL_SIZE,
            THREAD_POOL_SIZE,
            0L,
            TimeUnit.MILLISECONDS,
            new ArrayBlockingQueue<Runnable>(1000),
            namedThreadFactory,
            new ThreadPoolExecutor.AbortPolicy());

    private static DemoUser makeUser(int id){
        DemoUser demoUser = new DemoUser(id);
        String userName = "ruirui_"+id;
        demoUser.setName(userName);
        return demoUser;

    }

    private static class ProduceWorker implements Runnable{
        // 定义生产者消息数据
        private ProducerRecord<String,String> record;
        // 定义kafka的生产者
        private KafkaProducer<String,String> producer;

        public ProduceWorker(ProducerRecord<String, String> record,
                             KafkaProducer<String, String> producer) {
            this.record = record;
            this.producer = producer;
        }

        @Override
        public void run() {
            final String id = Thread.currentThread().getId()
                    +"-"+System.identityHashCode(producer);
            try {
                producer.send(record, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata,
                                             Exception exception) {
                        if(null!=exception){
                            exception.printStackTrace();
                        }
                        if(null!=metadata){
                            System.out.println(id+"|"
                                    +String.format("偏移量：%s,分区：%s",
                                    metadata.offset(),metadata.partition()));
                        }
                    }
                });
                System.out.println(id+":数据["+record+"]已发送。");
                countDownLatch.countDown();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        KafkaProducer<String,String> producer
                = new KafkaProducer<String, String>(
                KafkaConstants.producerConfig(StringSerializer.class,
                        StringSerializer.class));
        try{
            for (int i = 0; i < MSG_SIZE; i++) {
                DemoUser demoUser = makeUser(i);
                ProducerRecord<String,String> record
                        = new ProducerRecord<String,String>(
                        CommonConstant.REBALANCE_TOPIC,null,
                        System.currentTimeMillis(),
                        demoUser.getId()+"", demoUser.toString());
                executor.submit(new ProduceWorker(record,producer));
                Thread.sleep(600);
            }
            countDownLatch.await();

        }catch (Exception e){
            e.printStackTrace();
        }finally {
            producer.close();
            executor.shutdown();
        }
    }



}
